package doris;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaDataProducer {
    public static void main(String[] args) throws InterruptedException {
        String topic = "flink-doris-topic";
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "sjz01:9092,sjz02:9092,sjz03:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        while (true) {
            for (int i = 0; i < 100; i++) {
                String value = "user_" + i + ",activity_" + i + "," + System.currentTimeMillis();
                producer.send(new ProducerRecord<>(topic, Integer.toString(i), value));
            }
            Thread.sleep(1000 * 30);
        }


    }
}
